package com.atguigu.gmall.realtime.app.func;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.atguigu.gmall.realtime.beans.TableProcess;
import com.atguigu.gmall.realtime.common.GmallConfig;
import com.atguigu.gmall.realtime.utils.PhoenixUtil;
import org.apache.commons.lang3.StringUtils;
import org.apache.flink.api.common.functions.MapFunction;

public class CheckTableFunction implements MapFunction<String,String> {
    @Override
    public String map(String tableProcessJsonString) throws Exception {
        JSONObject tableProcessJsonObj = JSON.parseObject(tableProcessJsonString);
        String crudOptType = tableProcessJsonObj.getString("op"); //c: create ,r: retrieve   ,u: upate ,d: delete
        //查询初始化快照 op 是 r 操作
        TableProcess afterTableProcess = tableProcessJsonObj.getObject("after", TableProcess.class);

        if("dim".equals(afterTableProcess.getSinkType())){
            //op是  r  或者 c
            if(crudOptType.equals("r")||crudOptType.equals("c")){
                autoCreateTable(afterTableProcess);
            }else if(crudOptType.equals("u")){

            }
        }

        return tableProcessJsonString;
    }

    public  void  autoCreateTable(TableProcess tableProcess){
        //1 拼接create table语句
        // create table if not exists 库名.表名
        // (  字段1  varchar  primary key , 字段2 varchar ,...）
        // slat_buckets=xxx
        String databaseName= GmallConfig.DIM_SCHEMA;
        String tableName=tableProcess.getSinkTable();
        String pkCol = tableProcess.getSinkPk();
        if(pkCol == null){
            pkCol = "id";
        }
        StringBuilder colStringBuilder=new StringBuilder();
        String sinkColumns = tableProcess.getSinkColumns();
        String[] sinkColsArr = StringUtils.split(sinkColumns, ",");
        for (String col : sinkColsArr) {
            if(colStringBuilder.length()>0){
                colStringBuilder.append(",");
            }
            colStringBuilder.append(col+" varchar");
            if(col.equals(pkCol)){
                colStringBuilder.append(" primary key");
            }
        }
        String slat= tableProcess.getSinkExtend()!=null?tableProcess.getSinkExtend():"";

        String createTableSQL="create table if not exists "+databaseName+"."+tableName+
                "("+colStringBuilder.toString()+")"+slat;

        System.out.println("建表语句："+ createTableSQL);


        //2  把建表语句执行到phoenix
        PhoenixUtil.executeSQL(createTableSQL);


    }

    //练习： 调整维度字段
    public void addColumns(TableProcess beforeTableProcess,TableProcess afterTableProcess){
        //1  比较修改前后修改后对的sinkColumns  ,找出新增的字段
        //2  alter table xxxx  add  column  xxx
        //3  执行该alter语句。

    }
}
